package com.allwe.client.concurrent;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 多线程安全消费者 - 测试demo
 *
 * @Author: AllWe
 * @Date: 2024/09/27/12:34
 */
@Data
@Slf4j
public class ConcurrentConsumerWorker {
    /**
     * 消费线程数
     */
    private static final Integer THREAD_COUNT = 2;

    /**
     * 线程池 - 2个线程,别超过目标主题的分区数
     */
    private static ExecutorService executorService = Executors.newFixedThreadPool(THREAD_COUNT);


    public static void main(String[] args) {
        // 消费者配置
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "39.107.66.31:9092");
        properties.put("key.deserializer", StringDeserializer.class);
        properties.put("value.deserializer", StringDeserializer.class);
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // 从头开始消费

        for (Integer i = 0; i < THREAD_COUNT; i++) {
            executorService.submit(new ConcurrentConsumer(properties, "allwe01", "topic_6"));
        }
    }
}
